use std::{future::Future, pin::Pin, sync::Arc, time::Duration};

use futures_util::{stream::FuturesUnordered, StreamExt};
use lazy_static::lazy_static;
use redb::{ReadableTable, TableDefinition};
use ruma::{signatures::Ed25519KeyPair, DeviceId, UserId};
use tokio::time::sleep;
use tracing::debug;

use crate::{server::rooms, utils, Error, Result, DATABASE};

pub struct Service {
    services: Services,
    keypair: ruma::signatures::Ed25519KeyPair,
}

lazy_static! {
    pub static ref SERVICE: Arc<Service> = Arc::new(Service::new());
}

struct Services {
    state_cache: Arc<rooms::state_cache::Service>,
}

const GLOBAL_ATTR_DEFINITION: TableDefinition<&str, u64> = TableDefinition::new("current_count");

const COUNT_KEY: &str = "count";

impl Service {
    pub fn new() -> Self {
        let db = &DATABASE;
        let write_ctx = db.begin_write().expect("Database should be writable");
        write_ctx.open_table(GLOBAL_ATTR_DEFINITION).unwrap(); // create a table
        write_ctx.commit().expect("Database should be writable");
        Self {
            keypair: Self::load_keypair().expect("keypair should be loaded"),
            services: Services {
                state_cache: rooms::state_cache::SERVICE.clone(),
            },
        }
    }

    async fn worker(&self) -> Result<()> {
        for _ in 0..5 {
            sleep(Duration::from_secs(1)).await;
        }
        Ok(())
    }

    fn watch_fake<'a>(&'a self) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
        Box::pin(async move {
            // Tx is never destroyed
            self.worker().await.unwrap()
        })
    }

    pub async fn watch(&self, user_id: &UserId, device_id: &DeviceId) -> Result<()> {
        let userid_bytes = user_id.as_bytes().to_vec();
        let mut userid_prefix = userid_bytes.clone();
        userid_prefix.push(0xFF);

        let mut userdeviceid_prefix = userid_prefix.clone();
        userdeviceid_prefix.extend_from_slice(device_id.as_bytes());
        userdeviceid_prefix.push(0xFF);

        let mut futures = FuturesUnordered::new();

        futures.push(self.watch_fake());
        // Return when *any* user changed their key
        // TODO: only send for user they share a room with
        // futures.push(self.todeviceid_events.watch_prefix(&userdeviceid_prefix));

        // futures.push(self.userroomid_joined.watch_prefix(&userid_prefix));
        // futures.push(self.userroomid_invitestate.watch_prefix(&userid_prefix));
        // futures.push(self.userroomid_leftstate.watch_prefix(&userid_prefix));

        // futures.push(self.userroomid_highlightcount.watch_prefix(&userid_prefix));

        // // Events for rooms we are in
        // for room_id in self
        //     .services
        //     .state_cache
        //     .joined_rooms(user_id)
        //     .filter_map(Result::ok)
        // {
        //     let short_roomid = self
        //         .services
        //         .short
        //         .get_shortroomid(&room_id)
        //         .ok()
        //         .flatten()
        //         .expect("room exists")
        //         .to_be_bytes()
        //         .to_vec();

        //     let roomid_bytes = room_id.as_bytes().to_vec();
        //     let mut roomid_prefix = roomid_bytes.clone();
        //     roomid_prefix.push(0xFF);

        //     // PDUs
        //     futures.push(self.pduid_pdu.watch_prefix(&short_roomid));

        //     // Key changes
        //     futures.push(self.keychangeid_userid.watch_prefix(&roomid_prefix));

        //     // Room account data
        //     let mut roomuser_prefix = roomid_prefix.clone();
        //     roomuser_prefix.extend_from_slice(&userid_prefix);
        // }

        // let mut globaluserdata_prefix = vec![0xFF];
        // globaluserdata_prefix.extend_from_slice(&userid_prefix);

        // // More key changes (used when user is not joined to any rooms)
        // futures.push(self.keychangeid_userid.watch_prefix(&userid_prefix));

        // // One time keys
        // futures.push(self.userid_lastonetimekeyupdate.watch_prefix(&userid_bytes));

        // futures.push(Box::pin(async move {
        //     while self.services.server.running() {
        //         let _result = self.services.server.signal.subscribe().recv().await;
        //     }
        // }));

        // Wait until one of them finds something
        debug!(futures = futures.len(), "watch started");
        futures.next().await;
        debug!(futures = futures.len(), "watch finished");

        Ok(())
    }

    pub fn load_keypair() -> Result<Ed25519KeyPair> {
        let keypair_bytes = utils::generate_keypair();

        let mut parts = keypair_bytes.splitn(2, |&b| b == 0xFF);
        std::str::from_utf8(
            // 1. version
            parts
                .next()
                .expect("splitn always returns at least one element"),
        )
        .map(|s| s.to_owned())
        .map_err(|_| Error::bad_database("Invalid version bytes in keypair."))
        .and_then(|version| {
            // 2. key
            parts
                .next()
                .ok_or_else(|| Error::bad_database("Invalid keypair format in database."))
                .map(|key| (version, key))
        })
        .and_then(|(version, key)| {
            Ed25519KeyPair::from_der(key, version)
                .map_err(|_| Error::bad_database("Private or public keys are invalid."))
        })
    }

    pub fn current_count(&self) -> u64 {
        let db = &DATABASE;
        let read_ctx = db.begin_read().expect("Database should be readable");
        let count_table = read_ctx.open_table(GLOBAL_ATTR_DEFINITION).unwrap();
        let count = count_table
            .get(COUNT_KEY)
            .unwrap()
            .map(|v| v.value())
            .unwrap_or_default();
        count
    }

    pub fn next_count(&self) -> Result<u64> {
        let db = &DATABASE;
        let write_ctx = db.begin_write().expect("Database should be writable");
        let mut next_count = 0;
        {
            let mut count_table = write_ctx.open_table(GLOBAL_ATTR_DEFINITION).unwrap();
            let count = count_table
                .get(COUNT_KEY)
                .unwrap()
                .map(|v| v.value())
                .unwrap_or_default();
            next_count = count.checked_add(1).expect("Count should not overflow.");
            count_table
                .insert(COUNT_KEY, next_count)
                .map_err(|_| Error::bad_database("Failed to update count."))?;
        }
        write_ctx.commit().expect("Database should be writable");
        Ok(next_count)
    }

    pub fn keypair(&self) -> &ruma::signatures::Ed25519KeyPair {
        &self.keypair
    }

    pub fn allow_encryption(&self) -> bool {
        true
    }
}
